package org.test.mq.common;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.test.mq.common.exception.AceMQException;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;

/**
 * RocketMQ 消息发送者
 * 
 * @author tanjunyi
 * @time 2018-01-14
 */

public class AceMQProducer {
	public static final Logger LOGGER = LoggerFactory.getLogger(AceMQProducer.class);

	private DefaultMQProducer producer;
	private String groupName;
	private String namesrvAddr;

	public String getNamesrvAddr() {
		return namesrvAddr;
	}

	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}

	public String getGroupName() {
		return groupName;
	}

	public void setGroupName(String groupName) {
		this.groupName = groupName;
	}

	/**
	 * 初始化
	 * 
	 * @throws AceMQException
	 */
	public void init() throws AceMQException {
		this.producer = new DefaultMQProducer(this.groupName);
		this.producer.setNamesrvAddr(this.namesrvAddr);
		this.producer.setInstanceName("Producer");
		this.producer.setVipChannelEnabled(false);
		// 判断groupName是否为空
		if (StringUtils.isBlank(this.groupName)) {
			throw new AceMQException(" ======== ERROR:group name is blank==========");
		}
		// 判断namesrvAddr是否为空
		if (StringUtils.isBlank(this.namesrvAddr)) {
			throw new AceMQException("======== ERROR:namesrvAddr is blank======== ");
		}

		try {
			this.producer.start();
			System.out.println("producer is start!");
			LOGGER.info(String.format("producer is start!"));
		} catch (MQClientException e) {
			e.printStackTrace();
			LOGGER.info(String.format("producer is error !"));
			throw new AceMQException(e);
		}
	}

	/**
	 * 消息发送
	 * 
	 * @param topic
	 *            消息主题
	 * @param tags
	 *            消息标签
	 * @param keys
	 *            自定义key，可以通过key查询消息
	 * @param messageText
	 *            消息内容
	 * @return 返回消息发送结果
	 * @throws AceMQException
	 */
	public SendResult sendMessage(String topic, String tags, String keys, String messageText) throws AceMQException {
		Message message = new Message(topic, tags, keys, messageText.getBytes());
		try {
			SendResult sendResult = this.producer.send(message);
			return sendResult;
		} catch (Exception e) {
			LOGGER.info("send message error:" + e);
			throw new AceMQException(e);
		}

	}

	/**
	 * 关闭连接
	 * 
	 * @throws AceMQException
	 */
	public void shutdown() throws AceMQException {
		this.producer.shutdown();
	}
}
